package org.lzfto.flink.demo.business.flinkTest.flinkObjects.connector.file;

import java.util.HashSet;
import java.util.Set;

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.types.DataType;
import org.lzfto.flink.demo.business.flinkTest.base.record.ExcelOptionRecord;

/**
 * 自定义Excel连接器cust-excel
 * </br>负责从excel中读取数据到table中，暂时不支持把table的数据写入到excel中
 * </br>用于非常简单的操作：读取文件内容，删除文件（如果有必要)
 * </br>
 * @author lzfto
 * @created 2025年6月4日 13:26:59
 */
public class CustomExcelConnectorFactory implements DynamicTableSourceFactory {

	private static final ConfigOption<String> PATH = ConfigOptions.key("path").stringType().noDefaultValue();
	private static final ConfigOption<String> FILE_NAME = ConfigOptions.key("file-name").stringType().noDefaultValue();
	private static final ConfigOption<String> EXTENSION = ConfigOptions.key("extension").stringType().defaultValue("xlsx");
	private static final ConfigOption<String> EXTENSION_HYPHEN = ConfigOptions.key("extension-hyphen").stringType().defaultValue("-");
	/**
	 * 读取后删除文件
	 */
	private static final ConfigOption<Boolean> DELETE_AFTER_READ = ConfigOptions.key("delete-after-read").booleanType().defaultValue(false);
	private static final ConfigOption<Integer> START_ROW = ConfigOptions.key("start-row").intType().defaultValue(0);
	
    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
    	final FactoryUtil.TableFactoryHelper helper =
				FactoryUtil.createTableFactoryHelper(this, context);
		final ReadableConfig config = helper.getOptions();
		Boolean deleteAfterRead=config.get(DELETE_AFTER_READ);
		ExcelOptionRecord option=new ExcelOptionRecord(
				config.get(PATH),
				config.get(FILE_NAME),
				config.get(EXTENSION),
				config.get(EXTENSION_HYPHEN),
				deleteAfterRead,
				config.get(START_ROW)
				);
		DataType dataType = context.getPhysicalRowDataType();
		return new CustExcelTableSource(option, dataType);
    }

    @Override
    public String factoryIdentifier() {
        return "cust-excel";
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
    	Set<ConfigOption<?>> options = new HashSet<>();
		options.add(PATH);
		options.add(FILE_NAME);
		return options;
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
    	Set<ConfigOption<?>> options = new HashSet<>();
		options.add(DELETE_AFTER_READ);
		options.add(START_ROW);
		options.add(EXTENSION);
		options.add(EXTENSION_HYPHEN);
		return options;
    }

    @Override
    public Set<ConfigOption<?>> forwardOptions() {
        return DynamicTableSourceFactory.super.forwardOptions();
    }
}
